package com.maxbill.utils;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.maxbill.bean.JobBean;

/**
 * @功能 Hadoop Map/Reduce操作工具类
 * @作者 zuoshuai(MaxBill)
 */
public class JobsUtils {

	private final static String hdfsPath = "hdfs://192.168.1.10:9000";
	private final static String jobsPath = "hdfs://192.168.1.10:9001";

	/**
	 * @功能 获取HDFS的配置信息
	 * @作者 zuoshuai(MaxBill)
	 */
	public static Configuration getConfig() {
		Configuration config = new Configuration();
		config.set("fs.default.name", hdfsPath);
		config.set("mapred.job.tracker", jobsPath);
		return config;
	}

	/**
	 * @功能 获取HDFS的job配置信息
	 * @作者 zuoshuai(MaxBill)
	 */
	public static Job getJobs(String jobName, Object appClass, Object mapperClass, Object reduceClass) {
		Job job = null;
		try {
			job = Job.getInstance(getConfig());
			job.setJobName(jobName);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(IntWritable.class);
			job.setJarByClass((Class<?>) appClass);
			job.setMapperClass((Class<? extends Mapper>) mapperClass);
			job.setCombinerClass((Class<? extends Reducer>) reduceClass);
			job.setReducerClass((Class<? extends Reducer>) reduceClass);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return job;
	}

	/**
	 * @功能 执行作业
	 * @作者 zuoshuai(MaxBill)
	 */
	public static void runJob(JobBean jobBean) {
		try {
			Job job = getJobs(jobBean.getJobName(), jobBean.getAppClass(), jobBean.getMapperClass(),
					jobBean.getReduceClass());
			FileInputFormat.addInputPath(job, new Path(jobBean.getInputPath()));
			FileOutputFormat.setOutputPath(job, new Path(jobBean.getOutputPath()));
			System.exit(job.waitForCompletion(true) ? 0 : 1);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
